/**
 * 
 */
package stringreduce;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.regex.Pattern;

/**
 * @author timrobertson
 *
 */
public class StringMapReduce {
	static Pattern tab = Pattern.compile("\t");
	
	// hashmap that can return a sorted list of it's keys
	static class Collector extends HashMap<String, List<String>> {
		public Collector(int size) {
			super(size);
		}
		public void collect(String k, String v) {
			if (this.containsKey(k)) {
				this.get(k).add(v);
			} else {
				List<String> values = new LinkedList<String>();
				values.add(v);
				this.put(k, values);
			}
		}
		public List<String> sortedKeys() {
			List<String> k = new ArrayList<String>(this.keySet());
			Collections.sort(k);
			return k;
		}
	}
	
	static interface Mapper {
		public void map(String k, String v, Collector c);
	}
	
	static interface Reducer {
		public void reduce(String k, Collection<String> v);
	}
	
	public static void run(File input, Class<? extends Mapper> mapper, Class<? extends Reducer> reducer) throws Exception {
		// for transfer between map -> reduce
		Collector c = new Collector(1000000);
		
		// map
		Mapper m = mapper.newInstance();
		BufferedReader reader = new BufferedReader(new FileReader(input));
		
		String line;
		int count=0;
		int reportingCount=10000;
		while ((line=reader.readLine())!=null) {
			String[] parts = tab.split(line);
			m.map(parts[0], line, c);
			if (++count % reportingCount==0) {
				System.out.println("Mapped[" + count + "]");
			}
		}		
		reader.close();
		
		// reduce
		System.out.println("Starting sort and reduce");
		Reducer r = reducer.newInstance();
		for (String k : c.sortedKeys()) {
			r.reduce(k, c.get(k));
		}	
	}
	
	
	
	static class WordCount implements Mapper, Reducer {
		@Override
		public void map(String k, String v, Collector c) {
			for (String s : tab.split(v)) {
				c.collect(s, null);
			}
		}

		@Override
		public void reduce(String k, Collection<String> v) {
			System.out.println(k + ": " + v.size());
		}
	}
	
	public static void main(String[] args) {
		try {
			run(new File(args[0]), WordCount.class, WordCount.class);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
}
